package order

import (
	"context"
	"fmt"
	"github.com/sirupsen/logrus"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog/v2"
	cfg "order/pkg/config"
	"order/pkg/log"
	"order/pkg/utils"
	"path"
	"time"
)

const (
	CreateEvent = iota
	UpdateEvent
	DeleteEvent
)

type WrappEvent struct {
	event          int
	oldObj, newObj interface{}
}

type Controller struct {
	config    *cfg.Config
	logger    *logrus.Entry
	webhook   *WebHookServer
	informers informers.SharedInformerFactory
	stopCh    chan struct{}
}

func NewController(config *cfg.Config) *Controller {
	// 执行 k8s informer
	clientset, err := InitClientSet()
	Must(err)
	return &Controller{
		config:    config,
		webhook:   NewWebHookServer(config),
		informers: informers.NewSharedInformerFactoryWithOptions(clientset, 0),
		stopCh:    make(chan struct{}),
	}
}

func InitClientSet() (*kubernetes.Clientset, error) {
	userHome, err := utils.Home()
	if err != nil {
		userHome = "/"
	}
	kubeconfig := path.Join(userHome, clientcmd.RecommendedHomeDir, clientcmd.RecommendedFileName)
	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		panic(err)
	}
	return kubernetes.NewForConfig(config)
}

func (ctr *Controller) Start(ctx context.Context) error {
	log := log.InitLogger(ctr.config.LogLevel, ctr.config.NodeName)
	ctr.logger = log
	if ctr.config.WebHookServer {
		go ctr.webhook.StartWebhookServer(ctx)
	}
	if ctr.config.Controller {
		go ctr.syncPods(ctx)
		go ctr.syncPV(ctx)
	}
	select {
	case <-time.After(60 * time.Second):
		ctr.stopCh <- struct{}{}
	}
	return nil
}

func (ctr *Controller) syncPods(ctx context.Context) error {
	podsQ := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pods")
	podInformer := ctr.informers.Core().V1().Pods().Informer()
	podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			fmt.Println("add pod event...")
			podsQ.AddRateLimited(WrappEvent{CreateEvent, nil, obj})
		},
		UpdateFunc: func(oldObj, newObj interface{}) {
			fmt.Println("update pod event...")
			podsQ.AddRateLimited(WrappEvent{UpdateEvent, oldObj, newObj})
		},
		DeleteFunc: func(obj interface{}) {
			fmt.Println("delete pod event...")
			podsQ.AddRateLimited(WrappEvent{DeleteEvent, nil, obj})
		},
	})
	ctr.informers.Start(ctr.stopCh)
	defer ctr.informers.WaitForCacheSync(ctr.stopCh)
	processQueue := func() {
		for {
			obj, shutdown := podsQ.Get()
			if shutdown {
				return
			}
			eventItem := obj.(WrappEvent)
			pod, ok := eventItem.newObj.(*corev1.Pod)
			if !ok {
				klog.Errorf("Unexpected object in queue: %+v", eventItem)
				podsQ.Forget(obj)
				continue
			}
			podKey, err := cache.MetaNamespaceKeyFunc(pod)
			if err != nil {
				panic(err)
			}
			switch eventItem.event {
			case CreateEvent:
				fmt.Printf("Processing Create Pod %s status %s\n", podKey, pod.Status.Phase)
				// 此处增加 向rabbitMQ 发送消息的机制
			case UpdateEvent:
				// 此处可以查询状态
				fmt.Printf("Processing Update Pod %s status %s\n", podKey, pod.Status.Phase)
			case DeleteEvent:
				fmt.Printf("Processing Delete Pod %s status %s\n", podKey, pod.Status.Phase)
			}
			podsQ.Forget(obj)
		}
	}
	wait.Until(processQueue, time.Second, ctr.stopCh)
	return nil
}

func (ctr *Controller) syncPV(ctx context.Context) error {
	pvQ := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pv")
	pvInformer := ctr.informers.Core().V1().PersistentVolumes().Informer()
	pvInformer.AddEventHandler(
		&cache.ResourceEventHandlerFuncs{
			AddFunc: func(obj interface{}) {
				pv := obj.(*corev1.PersistentVolume)
				fmt.Printf("PV %s created\n", pv.Name)
				pvQ.AddRateLimited(WrappEvent{CreateEvent, nil, obj})
			},
			UpdateFunc: func(oldObj, newObj interface{}) {
				pv := newObj.(*corev1.PersistentVolume)
				fmt.Printf("PV %s updated\n", pv.Name)
				pvQ.AddRateLimited(WrappEvent{UpdateEvent, oldObj, newObj})
			},
			DeleteFunc: func(obj interface{}) {
				pv := obj.(*corev1.PersistentVolume)
				fmt.Printf("PV %s deleted\n", pv)
				pvQ.AddRateLimited(WrappEvent{DeleteEvent, nil, obj})
			},
		},
	)
	ctr.informers.Start(ctr.stopCh)
	defer ctr.informers.WaitForCacheSync(ctr.stopCh)
	processQueue := func() {
		for {
			obj, shutdown := pvQ.Get()
			if shutdown {
				return
			}
			eventItem := obj.(WrappEvent)
			pv, ok := eventItem.newObj.(*corev1.PersistentVolume)
			if !ok {
				klog.Errorf("Unexpected object in queue: %+v", eventItem)
				pvQ.Forget(obj)
				continue
			}
			switch eventItem.event {
			case CreateEvent:
				fmt.Printf("Processing Create PV %s status %s\n", pv.Name, pv.Status.Phase)
				// 此处增加 向rabbitMQ 发送消息的机制
			case UpdateEvent:
				// 此处可以查询状态
				fmt.Printf("Processing Update PV %s status %s\n", pv.Name, pv.Status.Phase)
			case DeleteEvent:
				fmt.Printf("Processing Delete PV %s status %s\n", pv.Name, pv.Status.Phase)
			}
			pvQ.Forget(obj)
		}
	}
	wait.Until(processQueue, time.Second, ctr.stopCh)
	return nil
}

func Must(e interface{}) {
	if e != nil {
		panic(e)
	}
}
